package org.pzy.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;

import java.io.IOException;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;

/**
 * @Author: DRevelation
 * @Date: 2022/11/3 15:35
 */
@Slf4j
public class ConsumerExample {
    public static void main(String[] args) throws ClientException, InterruptedException, IOException {
        String endpoint = "192.168.56.101:8081";

        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfiguration configuration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoint).build();

        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        String consumerGroup = "ConsumerGroup-20240320";
        String topic = "Topic-20240320";

        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(configuration)
                .setConsumerGroup(consumerGroup)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .setMessageListener(messageView -> {
                    log.info("Consumer Message:{}", messageView);
                    System.out.println("Consumer Success");
                    return ConsumeResult.SUCCESS;
                }).build();

        new CountDownLatch(1).await();
        pushConsumer.close();
    }
}
